package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.common.Constant;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/10/26 9:17
 */
public class FlinkSourceUtil {
    public static FlinkKafkaConsumer<String> getKafkaSource(String groupId,
                                                            String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("group.id", groupId);
        props.put("auto.offset.reset", "latest"); // 如果没有上次的消费记录, 则从最新开始消费. 如果有记录, 则从上次的位置开始消费
        props.put("isolation.level", "read_committed"); // 影响时效性
        
        return new FlinkKafkaConsumer<String>(
            topic,
            new SimpleStringSchema(),
            props
        );
    }
}
